package com.lvmama.java.rhino.etl.core;

import java.io.Serializable;

import org.apache.commons.lang.StringUtils;
import org.apache.spark.SparkConf;
import org.springframework.beans.factory.InitializingBean;

import com.lvmama.java.rhino.spark.core.ICallback;
import com.lvmama.java.rhino.spark.core.ISparkExecuting;
import com.lvmama.java.rhino.spark.core.ISparkShutdown;
import com.lvmama.java.rhino.spark.core.zookeeper.WatchExecutor;
import com.lvmama.java.rhino.spark.core.zookeeper.ZNodeEvent;
import com.lvmama.java.rhino.spark.core.zookeeper.ZNodeListener;
import com.lvmama.java.rhino.spark.utils.Constants;
import com.lvmama.java.rhino.spark.utils.ZookeeperUtils;

public abstract class AbstractSparkBaseTemplate implements ISparkExecuting, ISparkShutdown, Serializable, InitializingBean{
	private static final long serialVersionUID = -117718416014623074L;
	
	/** 线程名+日志作为一整行被临时存储到一行时的分隔符 */
	protected static final String THREAD_LOG_SPLIT = ";";
	/** 多条日志合并成一整行被临时存储到一行时的分隔符 */
	protected static final String LOG_SPLIT = ",";
	private static final String SPARK_ZOOKEEPER_ROOT = Constants.SPARK_ZOOKEEPER_ROOT;
	transient protected SparkConf sparkConf = null;
	transient protected ICallback callback = null;
	transient private WatchExecutor watchExecutor = null;
	@Override
	public void run(String[] args, ICallback callback) {
		this.callback = callback;
		String appName = args[0];
		init(appName, callback);
		// 创建Spark上下文
		sparkConf = getSparkConf(appName);
		run(args);
	}
	
	@Override
	public void afterPropertiesSet() throws Exception {
		if(!ZookeeperUtils.getInstance().exist(SPARK_ZOOKEEPER_ROOT)) {
			ZookeeperUtils.getInstance().create(SPARK_ZOOKEEPER_ROOT, "this is sparkstreaming root path");
		}
	}
	
	protected abstract void run(String[] args);
	
	private void init(String appName, final ICallback callback) {
		String path = SPARK_ZOOKEEPER_ROOT + "/" + appName;
		if (!ZookeeperUtils.getInstance().exist(path)) {
			ZookeeperUtils.getInstance().create(path, "sparkstreaming proceesor");
		}
		watchExecutor = new WatchExecutor(path, new ZNodeListener() {
			@Override
			public void process(ZNodeEvent event) {
				if("NodeDeleted".equals(event.getEventType())) {
					closing(-1);
				}
			}
			
			@Override
			public void closing(int rc) {
				shutdown(new String[]{});
				callback.call(null);
			}
		});
		watchExecutor.start();
	}
	
	protected SparkConf getSparkConf(String appName) {
		// 创建Spark上下文
		SparkConf sparkConf = new SparkConf().setAppName(appName)
				.set("spark.driver.allowMultipleContexts", "true")
				.set("spark.testing.memory", "2147480000000");
		// 从配置文件中读取，判断是否为本地debug模式
		String isDebugStr = Constants.getInstance().getValue("local.debug.run");
		if (StringUtils.isNotBlank(isDebugStr) && isDebugStr.equalsIgnoreCase("true")) {
			sparkConf.setMaster("local[2]");
		}
		return sparkConf;
	}
}
